package sdk.io.grpc.helloworld;

import static com.google.common.util.concurrent.MoreExecutors.directExecutor;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import com.google.common.base.Verify;
import com.google.common.base.VerifyException;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.rpc.DebugInfo;

import io.grpc.CallOptions;
import io.grpc.ClientCall;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Metadata;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.Status;
import io.grpc.protobuf.ProtoUtils;
import io.grpc.stub.StreamObserver;
import sdk.io.grpc.helloworld.proto.HelloGrpc;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloBlockingStub;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloFutureStub;
import sdk.io.grpc.helloworld.proto.HelloGrpc.HelloStub;
import sdk.io.grpc.helloworld.proto.HelloRequest;
import sdk.io.grpc.helloworld.proto.HelloResponse;

public class DetailErrorSample {
    private static final Metadata.Key<DebugInfo> DEBUG_INFO_TRAILER_KEY = ProtoUtils.keyForProto(DebugInfo.getDefaultInstance());

    private static final DebugInfo DEBUG_INFO = DebugInfo.newBuilder().addStackEntries("stack_entry_1").addStackEntries("stack_entry_2").addStackEntries("stack_entry_3")
            .setDetail("detailed error info.").build();

    private static final String DEBUG_DESC = "detailed error description";

    public static void main(String[] args) throws Exception {
        new DetailErrorSample().run();
    }

    private ManagedChannel channel;

    void run() throws Exception {
        Server server = ServerBuilder.forPort(0).addService(new HelloGrpc.HelloImplBase() {
            @Override
            public void call(HelloRequest request, StreamObserver<HelloResponse> responseObserver) {
                Metadata trailers = new Metadata();
                trailers.put(DEBUG_INFO_TRAILER_KEY, DEBUG_INFO);
                responseObserver.onError(Status.INTERNAL.withDescription(DEBUG_DESC).asRuntimeException(trailers));
            }
        }).build().start();
        channel = ManagedChannelBuilder.forAddress("localhost", server.getPort()).usePlaintext().build();

        blockingCall();
        futureCallDirect();
        futureCallCallback();
        asyncCall();
        advancedAsyncCall();

        channel.shutdown();
        server.shutdown();
        channel.awaitTermination(1, TimeUnit.SECONDS);
        server.awaitTermination();
    }

    static void verifyErrorReply(Throwable t) {
        Status status = Status.fromThrowable(t);
        Metadata trailers = Status.trailersFromThrowable(t);
        Verify.verify(status.getCode() == Status.Code.INTERNAL);
        Verify.verify(trailers.containsKey(DEBUG_INFO_TRAILER_KEY));
        Verify.verify(status.getDescription().equals(DEBUG_DESC));
        try {
            Verify.verify(trailers.get(DEBUG_INFO_TRAILER_KEY).equals(DEBUG_INFO));
        } catch (IllegalArgumentException e) {
            throw new VerifyException(e);
        }
    }

    void blockingCall() {
        HelloBlockingStub stub = HelloGrpc.newBlockingStub(channel);
        try {
            stub.call(HelloRequest.newBuilder().build());
        } catch (Exception e) {
            verifyErrorReply(e);
        }
    }

    void futureCallDirect() {
        HelloFutureStub stub = HelloGrpc.newFutureStub(channel);
        ListenableFuture<HelloResponse> response = stub.call(HelloRequest.newBuilder().build());

        try {
            response.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            verifyErrorReply(e.getCause());
        }
    }

    void futureCallCallback() {
        HelloFutureStub stub = HelloGrpc.newFutureStub(channel);
        ListenableFuture<HelloResponse> response = stub.call(HelloRequest.newBuilder().build());

        final CountDownLatch latch = new CountDownLatch(1);

        Futures.addCallback(response, new FutureCallback<HelloResponse>() {
            @Override
            public void onSuccess(@Nullable HelloResponse result) {
                // Won't be called, since the server in this example always fails.
            }

            @Override
            public void onFailure(Throwable t) {
                verifyErrorReply(t);
                latch.countDown();
            }
        }, directExecutor());

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }

    void asyncCall() {
        HelloStub stub = HelloGrpc.newStub(channel);
        HelloRequest request = HelloRequest.newBuilder().build();
        final CountDownLatch latch = new CountDownLatch(1);
        StreamObserver<HelloResponse> responseObserver = new StreamObserver<HelloResponse>() {

            @Override
            public void onNext(HelloResponse value) {
                // Won't be called.
            }

            @Override
            public void onError(Throwable t) {
                verifyErrorReply(t);
                latch.countDown();
            }

            @Override
            public void onCompleted() {
                // Won't be called, since the server in this example always fails.
            }
        };
        stub.call(request, responseObserver);

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }

    /**
     * This is more advanced and does not make use of the stub. You should not
     * normally need to do this, but here is how you would.
     */
    void advancedAsyncCall() {
        ClientCall<HelloRequest, HelloResponse> call = channel.newCall(HelloGrpc.getCallMethod(), CallOptions.DEFAULT);

        final CountDownLatch latch = new CountDownLatch(1);

        call.start(new ClientCall.Listener<HelloResponse>() {

            @Override
            public void onClose(Status status, Metadata trailers) {
                Verify.verify(status.getCode() == Status.Code.INTERNAL);
                Verify.verify(trailers.containsKey(DEBUG_INFO_TRAILER_KEY));
                try {
                    Verify.verify(trailers.get(DEBUG_INFO_TRAILER_KEY).equals(DEBUG_INFO));
                } catch (IllegalArgumentException e) {
                    throw new VerifyException(e);
                }

                latch.countDown();
            }
        }, new Metadata());

        call.sendMessage(HelloRequest.newBuilder().build());
        call.halfClose();

        if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
            throw new RuntimeException("timeout!");
        }
    }
}
